package com.nami.dispatcher.controller.middleware.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.google.protobuf.InvalidProtocolBufferException;
import com.nami.dispatcher.facade.qos.ConfirmMsgDeliveryFacade;
import com.nami.dispatcher.facade.qos.request.ConfirmMsgDeliveryCmd;
import com.nami.gate.external.basic.constants.UpTopicConst;
import com.nami.transport.proto.codec.DownWsMsg;
import com.nami.transport.proto.codec.MqTransMsg;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

@Slf4j
@Component
@AllArgsConstructor
@RocketMQMessageListener(topic = UpTopicConst.QOS_TOPIC,
    consumerGroup = UpTopicConst.QOS_CONSUMER_GROUP,
    messageModel = MessageModel.BROADCASTING, consumeMode = ConsumeMode.CONCURRENTLY)
public class QosConsumer implements RocketMQListener<MessageExt> {

    private final ConfirmMsgDeliveryFacade qosWriteAppService;

    /**
     * 目前为广播
     *
     * @param body
     */
    @Override
    public void onMessage(MessageExt body) {
        try {
            TimeInterval timer = DateUtil.timer();
            if (body.getBody() == null) {
                log.warn("[qos] message is empty");
                return;
            }

            ConfirmMsgDeliveryCmd cmd = parseQosCmd(body.getBody());
            if (cmd == null) {
                log.warn("[qos] message is empty");
                return;
            }
            qosWriteAppService.execute(cmd);

            log.info("[qos] qos处理完毕【userId={}, receiverId={}】size={} use={}ms", cmd.getSenderId(), cmd.getReceiverId(),
                cmd.getMsgIds().size(),
                timer.interval());
        } catch (Exception var2) {
            var2.printStackTrace();
            log.error("[qos] 出现错误：{}", var2);
        }

    }

    private ConfirmMsgDeliveryCmd parseQosCmd(byte[] body) {

        try {
            MqTransMsg.Gate2RouterPublishAckMsg ack = MqTransMsg.Gate2RouterPublishAckMsg.parseFrom(body);
            ConfirmMsgDeliveryCmd cmd = new ConfirmMsgDeliveryCmd();
            DownWsMsg.PublishAckMsgBody publishAckMsgBody = DownWsMsg.PublishAckMsgBody.parseFrom(ack.getData());
            String msgId = publishAckMsgBody.getMsgId();
            String[] split = msgId.split(",");

            DownWsMsg.SimplifiedDownMsg modifiedMsg = publishAckMsgBody.getModifiedMsg();
            String msgContent = modifiedMsg.getMsgContent().toStringUtf8();
            if(msgContent.equals("c2c")) {
                cmd.setConverType("c2c");
            } else if (msgContent.equals("c2g")) {
                cmd.setConverType("c2g");
            }
            Set<String> list = Arrays.stream(split).collect(Collectors.toSet()) ;
            cmd.setMsgIds(list);
            cmd.setSenderId(ack.getUserId());
            cmd.setInstanceId(ack.getInstanceId());

            return cmd;
        } catch (InvalidProtocolBufferException e) {
            throw new RuntimeException(e);
        }
//        ConfirmMsgDeliveryCmd cmd = Serializer.prototype(packet.getSerial()).deserialize(packet.getBody(), ConfirmMsgDeliveryCmd.class);


    }


}
